package cn.itcast.streaming.window.function;

import cn.itcast.streaming.entity.ElectricFenceModel;
import com.google.common.collect.Lists;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;

/**
 * Author itcast
 * Date 2021/9/26 15:02
 * Desc 实现每个窗口内90s在电子栅栏内还是在电子栅栏外
 * 如果是上个状态在电子栅栏外，当前状态在电子栅栏内，进栅栏，否则是出栅栏，将进或出栅栏的时间写入到
 * 开发步骤：
 * 1.open方法中读取上个窗口内的状态，车辆在电子栅栏内还是电子栅栏外
 * 2.apply方法中，将时间或者在栅栏内还是栅栏外写入到对象中
 */
public class ElectricFenceWindowFunction extends RichWindowFunction<ElectricFenceModel, ElectricFenceModel, String, TimeWindow> {
    //创建logger
    private static final Logger logger = LoggerFactory.getLogger(ElectricFenceWindowFunction.class);
    //1.定义存储历史电子围栏数据的state，<vin，是否在电子围栏内0:内，1:外> MapState<String, Integer>
    MapState<String, Byte> lastState = null;

    //2.重写open方法
    @Override
    public void open(Configuration parameters) throws Exception {
        //2.1 定义mapState的描述器，就是定义state的结构
        MapStateDescriptor<String, Byte> lastStateDesc = new MapStateDescriptor<>("lastState", String.class, Byte.class);
        //2.2 获取 parameterTool，用来读取配置文件参数
        //2.3 读取状态的超时时间 "vehicle.state.last.period" ,构建ttl设置更新类型和状态可见
        //2.4 设置状态描述 StateTtlConfig，开启生命周期时间
        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.seconds(180))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        //设置state的ttl
        lastStateDesc.enableTimeToLive(stateTtlConfig);
        //2.5 获取map状态
        lastState = getRuntimeContext().getMapState(lastStateDesc);
    }

    @Override
    public void apply(String key, TimeWindow window, Iterable<ElectricFenceModel> input, Collector<ElectricFenceModel> out) throws Exception {
        //1.创建返回对象
        ElectricFenceModel electricFenceModel = new ElectricFenceModel();
        //2.对窗口内的数据进行排序
        ArrayList<ElectricFenceModel> electricFenceModelList = Lists.newArrayList(input);
        Collections.sort(electricFenceModelList);
        //3.从 state 中获取车辆vin对应的上一次窗口电子围栏lastStateValue标记（车辆上一次窗口是否在电子围栏中）0：电子围栏内 1：电子围栏外
        Byte lastStateValue = lastState.get(key);
        //4.如果上次状态为空，初始化赋值
        if (lastStateValue == null) {
            lastStateValue = -1;
        }
        //5.判断当前处于电子围栏内还是电子围栏外
        //5.1.定义当前车辆电子围栏圈内出现的次数
        long inElectricFence = electricFenceModelList.stream()
                .filter(model -> model.getNowStatus() == 0)
                .count();
        //5.2.定义当前车辆电子围栏圈外出现的次数
        long outElectricFence = electricFenceModelList.stream()
                .filter(model -> model.getNowStatus() == 1)
                .count();
        //6.定义当前窗口的电子围栏状态,默认让当前状态为1，在电子栅栏外
        byte currentState = 1;
        //7. 90s内车辆出现在电子围栏内的次数多于出现在电子围栏外的次数，则认为当前处于电子围栏内
        if (inElectricFence >= outElectricFence) {
            currentState = 0;
        }
        //8. 将当前窗口的电子围栏状态写入到 state 中，供下次判断
        lastState.put(key,currentState);
        //9.如果当前电子围栏状态与上一次电子围栏状态不同
        //9.1.如果上一次窗口处于电子围栏外，而本次是电子围栏内，则将进入电子围栏的时间写入到数据库中
        if(lastStateValue==1 && currentState==0) {
            //9.1.1.过滤出来状态为0的第一条数据
            ElectricFenceModel firstInEleModel = electricFenceModelList.stream()
                    .filter(model -> model.getNowStatus() == 0).findFirst().get();
            //9.1.2.拷贝属性给 electricFenceModel 并将进入终端时间赋值，并且将状态告警字段赋值为1 0:出围栏 1:进围栏，将数据collect返回
            BeanUtils.copyProperties(electricFenceModel,firstInEleModel);
            electricFenceModel.setInEleTime(firstInEleModel.getGpsTime());
            electricFenceModel.setStatusAlarm(1);
            out.collect(electricFenceModel);
        }else if(lastStateValue==0 && currentState ==1) {
            //9.2.如果上一次窗口处于电子围栏内，而本次是电子围栏外，则将出电子围栏的时间写入到数据库中
            //9.2.1.过滤出来状态为1的第一条数据|
            ElectricFenceModel firstOutEleModel = electricFenceModelList.stream()
                    .filter(model -> model.getNowStatus() == 1).findFirst().get();
            //9.2.2.拷贝属性给 electricFenceModel 并将出终端时间赋值，并且将状态告警 0:出围栏 1:进围栏，将数据collect返回
            BeanUtils.copyProperties(electricFenceModel,firstOutEleModel);
            electricFenceModel.setInEleTime(firstOutEleModel.getGpsTime());
            electricFenceModel.setStatusAlarm(0);
            out.collect(electricFenceModel);
        }
    }
}
